package com.viknix.threadpool.manager.client.core;

import cn.hutool.json.JSONUtil;
import com.viknix.threadpool.manager.client.annotations.ThreadPoolManager;
import com.viknix.threadpool.manager.client.builder.ThreadPoolBuilder;
import com.viknix.threadpool.manager.client.config.ApplicationContextHolder;
import com.viknix.threadpool.manager.client.consistency.CacheData;
import com.viknix.threadpool.manager.client.enums.BlockingQueueEnum;
import com.viknix.threadpool.manager.client.enums.RejectedTypeEnum;
import com.viknix.threadpool.manager.client.global.GlobalThreadPoolManage;
import com.viknix.threadpool.manager.client.http.HttpAgent;
import com.viknix.threadpool.manager.common.model.PoolParameterInfo;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import com.viknix.threadpool.manager.client.http.HttpSimpleClient.HttpResult;

import java.net.HttpURLConnection;
import java.util.*;
import java.util.concurrent.*;

import static com.viknix.threadpool.manager.common.constant.Constants.*;

/**
 * @Author: Dongqi
 * @Date: 2021/10/24 12:26
 * @Version 1.0
 * @Description: 动态线程池后置处理器
 */
@Slf4j
@AllArgsConstructor
public class DynamicThreadPoolPostProcessor implements BeanPostProcessor {

    private static final String DYNAMIC_REFRESH_CONFIG_THREAD = "dynamic-refresh-config-thread";

    private String projectId;

    private String namespace;

    private final HttpAgent agent;

    /**
     * 线程池行为对象
     */
    private final ThreadPoolOperation threadPoolOperation;

    public DynamicThreadPoolPostProcessor(String projectId, String namespace, HttpAgent agent, ConfigService configService) {
        this.projectId = projectId;
        this.namespace = namespace;
        this.agent = agent;
        this.threadPoolOperation = new ThreadPoolOperation(namespace, projectId, configService);
    }

    /**
     * 如果服务端刷新了动态线程池的配置，那么监听器会通过该线程池动态刷新内存中缓存的MD5值以及被更新的线程池的参数
     */
    private final ExecutorService refreshExecutor = ThreadPoolBuilder.builder()
            .setCorePoolSize(4)
            .setMaxPoolSize(4)
            .keepAliveTime(0L, TimeUnit.MILLISECONDS)
            .setBlockingQueueCode(BlockingQueueEnum.LINKED_BLOCKING_QUEUE.getCode())
            .setThreadNamePrefix(DYNAMIC_REFRESH_CONFIG_THREAD)
            // 丢弃队列最前面的任务，然后重新提交被拒绝的任务
            .setRejectedHandlerCode(RejectedTypeEnum.DISCARD_OLDEST_POLICY.getCode())
            .build();

    /**
     * 1、从服务端拉去配置信息，通过 namespace + projectId + threadPoolId 确定线程池配置；
     * 2、如果服务端有配置信息，服务端返回配置信息，客户端根据服务端的配置信息重新刷新动态线程池配置，如果服务端没有该线程池的配置信息，那么会向服务端注册一条该线程池的配置信息；
     * 3、将该线程池以及线程池参数注册到 全局管理器 {@link com.viknix.threadpool.manager.client.global.GlobalThreadPoolManage} 中；
     * 4、为该线程池注册监听器，并且将 listener 和该线程池配置的md5信息 封装成为 {@link CacheData} 对象；
     * 5、通过长轮询为 内存仓库中的所有监听器注册监听，如果服务端配置发生变化，那么将通过该线程池的监听器动态刷新当前服务中的线程池配置。
     *
     * @param bean     当前 ThreadFactory#getBean 的对象
     * @param beanName 当前 ThreadFactory#getBean 的对象的名称
     * @return 当前 ThreadFactory#getBean 的对象
     * @throws BeansException
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

        // 第一种方式：
        // 如果当前的bean是 DynamicThreadPoolExecutor
        if (bean instanceof DynamicThreadPoolExecutor) {
            // 看看当前 bean 上有没有 @ThreadPoolManager 注解
            var dynamicThreadPool = ApplicationContextHolder.findAnnotationOnBean(beanName, ThreadPoolManager.class);
            // 如果没有 @ThreadPoolManager 注解直接返回
            if (Objects.isNull(dynamicThreadPool)) {
                return bean;
            }
            // 如果有转为 DynamicThreadPoolExecutor 类型
            var dynamicExecutor = (DynamicThreadPoolExecutor) bean;
            // 将线程池包装成为 DynamicThreadPoolWrap
            var dynamicThreadPoolWrap = new DynamicThreadPoolWrap(dynamicExecutor.getThreadPoolId(), dynamicExecutor);
            // 根据服务端配置刷新线程池参数，并且为该线程池注册监听
            return registerAndSubscribe(dynamicThreadPoolWrap);
            // 第二种方式：
        } else if (bean instanceof DynamicThreadPoolWrap) {
            var dynamicThreadPoolWrap = (DynamicThreadPoolWrap) bean;
            // 根据服务端配置刷新线程池参数，并且为该线程池注册监听
            registerAndSubscribe(dynamicThreadPoolWrap);
        }
        return bean;
    }

    /**
     * 根据服务端配置刷新线程池参数，并且为该线程池注册监听
     *
     * @param dynamicThreadPoolWrap 线程池包装对象
     * @return 已经与服务端同步配置后的 DynamicThreadPoolExecutor 对象
     */
    protected DynamicThreadPoolExecutor registerAndSubscribe(DynamicThreadPoolWrap dynamicThreadPoolWrap) {
        // 从服务端拉取线程池配置，并且将线程池和线程池参数注册到全局管理器中
        DynamicThreadPoolExecutor dynamicThreadPoolExecutor = fillPoolAndRegister(dynamicThreadPoolWrap);
        // 订阅远端线程池配置：将 cacheData 绑定到 cacheMap，Listener 绑定到 cacheData
        subscribeConfig(dynamicThreadPoolWrap);
        return dynamicThreadPoolExecutor;
    }

    /**
     * 从服务端拉取线程池配置，并且将线程池和线程池参数注册到全局管理器中
     * 如果当前线程池参数和服务端线程池参数不一致，那么会重新创建线程池替换当前线程池
     *
     * @param dynamicThreadPoolWrap 线程池包装对象
     * @return 已经与服务端同步配置后的 DynamicThreadPoolExecutor 对象
     */
    protected DynamicThreadPoolExecutor fillPoolAndRegister(DynamicThreadPoolWrap dynamicThreadPoolWrap) {
        String poolId = dynamicThreadPoolWrap.getThreadPoolId();
        // 封装http请求参数
        List<String> params = new ArrayList<String>();
        // 获取线程池ID
        params.add(POOL_ID);
        params.add(poolId);
        // 获取项目ID
        params.add(ITEM);
        params.add(projectId);
        // 获取命名空间
        params.add(NAMESPACE);
        params.add(namespace);
        HttpResult result;
        boolean isSubscribe = false;
        DynamicThreadPoolExecutor poolExecutor = null;
        PoolParameterInfo poolParameter = null;

        try {
            // 向服务端发送Get请求 路径：/tpm/v1/configs
            // 根据 线程池ID、项目ID、命名空间 从服务端查询配置信息
            result = agent.httpGet(CONFIG_CONTROLLER_PATH, null, params, ENCODE, 3000L);
            if (HttpURLConnection.HTTP_OK == result.code && result.content != null) {
                log.info("[{}] [Init pool] ok, poolId={}, item={}, namespace={}, config={}", agent.getName(), poolId,
                        projectId, namespace, result.content);
                // 如果接口有返回值，根据配置信息重新构建 DynamicThreadPoolExecutor
                poolParameter = JSONUtil.toBean(result.content, PoolParameterInfo.class);
                dynamicThreadPoolWrap.getThreadPool().shutdown();
                // 使用相关参数创建线程池
                poolExecutor = (DynamicThreadPoolExecutor) ThreadPoolBuilder.builder()
                        .setDynamicThreadPool(true)
                        .setBlockingQueueCode(poolParameter.getQueueType())
                        .setCapacity(poolParameter.getCapacity())
                        .setThreadPoolId(poolId)
                        .setThreadNamePrefix(poolId)
                        .setCorePoolSize(poolParameter.getCorePoolSize())
                        .setMaxPoolSize(poolParameter.getMaxPoolSize())
                        .keepAliveTime(poolParameter.getKeepAliveTime(), TimeUnit.SECONDS)
                        .setRejectedHandlerCode(poolParameter.getRejectedType())
                        .alarmConfig(poolParameter.alert(), poolParameter.getCapacityAlert(), poolParameter.getLivenessAlert())
                        .build();
                dynamicThreadPoolWrap.setThreadPool(poolExecutor);
                isSubscribe = true;
            } else {
                log.error("[{}] [Init pool] fail, poolId={}, item={}, namespace={}, config={}", agent.getName(), poolId,
                        projectId, namespace, result.content);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("[{}] [Init pool] exception, poolId={}, item={}, msg={}", agent.getName(), poolId,
                    projectId, e.toString());
            poolExecutor = dynamicThreadPoolWrap.getThreadPool() != null ?
                    dynamicThreadPoolWrap.getThreadPool() : ThreadPoolBuilder.builder().defaultThreadPool(poolId);
            dynamicThreadPoolWrap.setThreadPool(poolExecutor);
            return poolExecutor;
        } finally {
            // 如果当前dynamicThreadPoolWrap中的pool属性为空，那么构建一个默认的线程池给他
            if (Objects.isNull(dynamicThreadPoolWrap.getThreadPool())) {
                dynamicThreadPoolWrap.setThreadPool(ThreadPoolBuilder.builder().defaultThreadPool(poolId));
            }
            // 设置是否订阅远端线程池配置
            // 如果这个值是false，那么就不会绑定到 cacheMap 上，不会向服务端注册监听
            dynamicThreadPoolWrap.setSubscribeFlag(isSubscribe);
            // 将线程池和线程池参数注册到内存Map中进行全局管理
            GlobalThreadPoolManage.register(dynamicThreadPoolWrap.getThreadPoolId(), poolParameter, dynamicThreadPoolWrap);
        }
        return poolExecutor;
    }

    /**
     * 为该线程池注册监听器，并且将 listener 和该线程池配置的md5信息 封装成为 {@link CacheData} 对象
     *
     * @param dynamicThreadPoolWrap 线程池包装对象
     */
    protected void subscribeConfig(DynamicThreadPoolWrap dynamicThreadPoolWrap) {
        // 是否需要订阅远端线程池配置
        if (dynamicThreadPoolWrap.isSubscribeFlag()) {
            // 通过 threadPoolOperation 订阅远端线程池配置
            threadPoolOperation.subscribeConfig(dynamicThreadPoolWrap.getThreadPoolId(), refreshExecutor, config -> ThreadPoolDynamicRefresh.refreshDynamicPool(config, agent));
        }
    }
}
